package com.wpen.unit.core.service.impl;

import java.util.ArrayList;
import java.util.Date;
import java.util.HashMap;
import java.util.HashSet;
import java.util.List;
import java.util.Map;
import java.util.Set;
import java.util.concurrent.Callable;
import java.util.concurrent.FutureTask;
import java.util.concurrent.LinkedBlockingQueue;
import java.util.concurrent.ThreadPoolExecutor;
import java.util.concurrent.TimeUnit;

import com.wpen.unit.core.service.DependonScheduleHandler;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.scheduling.concurrent.CustomizableThreadFactory;
import org.springframework.stereotype.Service;
import org.springframework.transaction.annotation.Propagation;
import org.springframework.transaction.annotation.Transactional;

import com.wpen.unit.AppVariables;
import com.wpen.unit.core.cache.CacheDatas;
import com.wpen.unit.core.cache.CacheTaskJobDpdon;
import com.wpen.unit.core.ddo.DependonCheckDo;
import com.wpen.unit.core.ddo.DependonCheckDo.Status;
import com.wpen.unit.core.service.CommunityHandler;
import com.wpen.unit.core.service.NoticeHelper;
import com.wpen.unit.core.service.TriggerPoolHelper;
import com.wpen.unit.enums.DependType;
import com.wpen.unit.enums.ExecStatus;
import com.wpen.unit.enums.YN;
import com.wpen.unit.infra.mybatis.job.po.BatchInfoPo;
import com.wpen.unit.infra.mybatis.job.po.DependonPo;
import com.wpen.unit.infra.mybatis.job.po.JobExec1LogPo;
import com.wpen.unit.infra.mybatis.job.po.JobInfoPo;
import com.wpen.unit.infra.mybatis.job.po.LockPo;
import com.wpen.unit.infra.mybatis.job.po.TaskExecLogPo;
import com.wpen.unit.infra.mybatis.job.po.TaskInfoPo;
import com.wpen.unit.infra.mybatis.job.po.TaskPlanPo;
import com.wpen.unit.infra.repository.job.JobExec1LogRepository;
import com.wpen.unit.infra.repository.job.JobInfoRepository;
import com.wpen.unit.infra.repository.job.LockRepository;
import com.wpen.unit.infra.repository.job.TaskExecLogRepository;
import com.wpen.unit.infra.repository.job.TaskPlanRepository;

import cn.hutool.core.collection.ConcurrentHashSet;
import lombok.extern.slf4j.Slf4j;

/**
 *
 *
 * @author Wang Peng
 * @date 2022年12月21日
 * 
 */
@Service
@Slf4j
public class DependonScheduleHandlerImpl implements DependonScheduleHandler {
    private final static String LOCK_NAME = "unit_depend_on_schedule";

    @Autowired
    private LockRepository lockRepository;
    @Autowired
    private TaskPlanRepository taskPlanRepository;
    @Autowired
    private TriggerPoolHelper triggerPoolHelper;
    @Autowired
    private TaskExecLogRepository taskExecLogRepository;
    @Autowired
    private JobExec1LogRepository jobExec1LogRepository;
    @Autowired
    private JobInfoRepository jobInfoRepository;
    @Autowired
    private CacheTaskJobDpdon cacheTjDpdon;
    @Autowired
    private CacheDatas cacheDatas;
    @Autowired
    private CommunityHandler communityHandler;
    @Autowired
    private NoticeHelper noticeHelper;

    /* key=batchid */
    private static Set<Long> runningBatch = new ConcurrentHashSet<>();

    private static ThreadPoolExecutor taskExecutorPool = null;

    static {
        if (taskExecutorPool == null) {
            taskExecutorPool = new ThreadPoolExecutor(4, 8, 2, TimeUnit.MINUTES, new LinkedBlockingQueue<Runnable>(16),
                    new CustomizableThreadFactory("job-executor-pool-"), new ThreadPoolExecutor.CallerRunsPolicy());
        }
    }

    @Override
    @Transactional(propagation = Propagation.REQUIRES_NEW)
    public void scheduleHandle() {
        // 锁表
        LockPo lockPo = lockRepository.queryForUpdate(LOCK_NAME);
        if (lockPo == null) {
            lockPo = new LockPo();
            lockPo.setLockName(LOCK_NAME);
            lockRepository.insert(lockPo);
            lockPo = lockRepository.queryForUpdate(LOCK_NAME);
        }

        // 延迟5秒一次
        // 查询,,,只查询近2天的，，2天前的不处理
        List<TaskPlanPo> taskPlanVaild0s = taskPlanRepository.queryAllVaild0();
        if (taskPlanVaild0s == null || taskPlanVaild0s.isEmpty()) {
            return;
        }
        for (TaskPlanPo taskPlanPo : taskPlanVaild0s) {
            runningBatch.add(taskPlanPo.getBatchExecLogId());
        }
        log.info("Running Batch >>>>>>>>>>" + runningBatch);
        List<FutureTask<Boolean>> listFutureTasks = new ArrayList<>();
        for (Long batchLogId : runningBatch) {
            Callable<Boolean> runner = new Callable<Boolean>() {

                @Override
                public Boolean call() throws Exception {
                    try {
                        // 查询tasklog
                        List<TaskExecLogPo> taskExecLogPos = taskExecLogRepository.queryListByBatchId(batchLogId, null);
                        if (taskExecLogPos == null || taskExecLogPos.isEmpty()) {
                            return true;
                        }
                        // 存储所有的execLog
                        Map<Long, TaskExecLogPo> mapTaskLogs = new HashMap<>();
                        // 配置信息，通过code 查找
                        Map<String, TaskInfoPo> mapTaskInfos = new HashMap<>();
                        Map<String, Map<String, JobInfoPo>> mapJobInfos = new HashMap<>();
                        // task判断是否完成，不能根据某一个。
                        // 统计运行中的running
                        List<Long> taskRunIds = new ArrayList<>();
                        // key=bbtcodes - batchTime
                        Map<String, List<TaskExecLogPo>> taskLogsGroupCodes = new HashMap<>();
                        Map<String, Boolean> taskLogsGroupBbtStatus = new HashMap<>();

                        for (TaskExecLogPo execLogPo : taskExecLogPos) {
                            mapTaskLogs.put(execLogPo.getId(), execLogPo);
                            String taskInfoKey = execLogPo.getBizCode() + "-" + execLogPo.getBatchCode() + "-"
                                    + execLogPo.getTaskCode();
                            String bbtCodeAndtime = taskInfoKey + "-" + execLogPo.getBatchTriggerTime();
                            if (taskLogsGroupCodes.containsKey(bbtCodeAndtime)) {
                                taskLogsGroupCodes.get(bbtCodeAndtime).add(execLogPo);
                            } else {
                                List<TaskExecLogPo> taskLogs = new ArrayList<>();
                                taskLogs.add(execLogPo);
                                taskLogsGroupCodes.put(bbtCodeAndtime, taskLogs);
                            }
                            if (!taskLogsGroupBbtStatus.containsKey(bbtCodeAndtime)) {
                                taskLogsGroupBbtStatus.put(bbtCodeAndtime, true);
                            }
                            // 后面值关注运行中的。。task表没有异常，只有运行中、成功
                            if (ExecStatus.Running.value().equals(execLogPo.getExecStatus())) {
                                taskRunIds.add(execLogPo.getId());
                                if (taskLogsGroupBbtStatus.get(bbtCodeAndtime)) {
                                    taskLogsGroupBbtStatus.put(bbtCodeAndtime, false);
                                }
                            }
                            // 根据code加载原来info信息
                            if (!mapTaskInfos.containsKey(taskInfoKey)) {
                                // 查询task配置信息
                                TaskInfoPo taskInfoPo = cacheDatas.getTaskInfoByCode(execLogPo.getBizCode(),
                                        execLogPo.getBatchCode(), execLogPo.getTaskCode());
                                mapTaskInfos.put(taskInfoKey, taskInfoPo);
                            }
                            if (!mapJobInfos.containsKey(taskInfoKey)) {
                                // 查询task所有作业
                                List<JobInfoPo> jobInfoPos = jobInfoRepository.queryListByTask(execLogPo.getBizCode(),
                                        execLogPo.getBatchCode(), execLogPo.getTaskCode());
                                if (jobInfoPos != null) {
                                    for (JobInfoPo jobPo : jobInfoPos) {
                                        Map<String, JobInfoPo> mapJob = new HashMap<>();
                                        mapJob.put(taskInfoKey + "-" + jobPo.getJobCode(), jobPo);
                                        mapJobInfos.put(taskInfoKey, mapJob);
                                    }
                                }
                            }
                        }
                        if (taskRunIds == null || taskRunIds.isEmpty()) {
                            // 运行中的task都完成了，分析一下依赖是否还有没有执行的
                            boolean allTaskFinish = true;
                            for (TaskExecLogPo execLogPo : taskExecLogPos) {
                                String bizCode = execLogPo.getBizCode();
                                String batchCode = execLogPo.getBatchCode();
                                String taskCode = execLogPo.getTaskCode();
                                // task下层job全部都完成了,,,,,，去读取task后面有没有依赖
                                DependonCheckDo dependonCheckDo = loopTaskDpdons(bizCode, batchCode, taskCode,
                                        taskLogsGroupBbtStatus, execLogPo.getBatchTriggerTime(),
                                        execLogPo.getTaskTriggerTime(), execLogPo.getExecDate());
                                if (Status.Finished != dependonCheckDo.getStatus()) {
                                    allTaskFinish = false;
                                }
                            }
                            // 如果完成，清理plan数据，修改状态等，，顺便在处理batch依赖
                            if (allTaskFinish) {
                                log.info("Batch Finished >>>>>>>>>>>>>>>>" + batchLogId);
                                // 更新当前batch状态为 完成
                                communityHandler.checkTaskFinished(batchLogId);
                                runningBatch.remove(batchLogId);
                            }
                            return true;
                        }
                        // 存在运行中的任务
                        runningCheck(taskRunIds, mapTaskLogs, mapTaskInfos);
                        return true;
                    } catch (Exception e) {
                        log.error(e.getMessage(), e);
                        return false;
                    }
                }

                private DependonCheckDo loopTaskDpdons(String bizCode, String batchCode, String taskCode,
                        Map<String, Boolean> taskLogsGroupBbtStatus, Long batchTriggerTime, Long taskTriggerTime,
                        String acctDate) {
                    DependonCheckDo checkDo = new DependonCheckDo();
                    List<DependonPo> dependonPos = cacheTjDpdon.getDpdsByParent(bizCode, batchCode, taskCode,
                            AppVariables.DEPENDON_PHD, DependType.Task);
                    Set<String> notRun = new HashSet<>();

                    if (dependonPos != null && !dependonPos.isEmpty()) {
                        for (DependonPo dpdPo : dependonPos) {
                            BatchInfoPo batchInfoPo = cacheDatas.getBatchInfoByCode(bizCode, batchCode);
                            String taskKeyDpd = dpdPo.getBizCode() + "-" + dpdPo.getBatchCode() + "-"
                                    + dpdPo.getTaskCode();
                            String bbtCatDpd = taskKeyDpd + "-" + batchTriggerTime;
                            if (!taskLogsGroupBbtStatus.containsKey(bbtCatDpd)) {
                                // 不包含，没有触发呢
                                // 先判断他的父节点是否都完成了
                                List<DependonPo> dpdInners = cacheTjDpdon.getDpdsByOwn(bizCode, batchCode,
                                        dpdPo.getTaskCode(), AppVariables.DEPENDON_PHD, DependType.Task);
                                boolean parentAllFinished = true;
                                for (DependonPo checkDpdParent : dpdInners) {
                                    String taskParentDpd = checkDpdParent.getDpdonBizCode() + "-"
                                            + checkDpdParent.getDpdonBatchCode() + "-"
                                            + checkDpdParent.getDpdonTaskCode();
                                    String bbtCatParentDpd = taskParentDpd + "-" + batchTriggerTime;
                                    if (!taskLogsGroupBbtStatus.containsKey(bbtCatParentDpd)) {
                                        parentAllFinished = false;
                                    }
                                }
                                if (parentAllFinished) {
                                    // 检测一次，taskplan 有没有记录
                                    List<TaskPlanPo> taskPlanPos = taskPlanRepository.queryListByCodes(bizCode,
                                            batchCode, dpdPo.getTaskCode(), batchTriggerTime);
                                    if (taskPlanPos == null || taskPlanPos.isEmpty()) {
                                        triggerPoolHelper.triggerDependon(batchLogId, dpdPo, batchInfoPo,
                                                batchTriggerTime, taskTriggerTime, acctDate);
                                        notRun.add(dpdPo.getTaskCode());
                                    }
                                }
                            } else {
                                // dpd已经完成，查询他还没有有其他
                                DependonCheckDo checkSubDo = loopTaskDpdons(bizCode, batchCode, dpdPo.getTaskCode(),
                                        taskLogsGroupBbtStatus, batchTriggerTime, taskTriggerTime, acctDate);
                                if (Status.Trigger == checkSubDo.getStatus()) {
                                    notRun.addAll(checkSubDo.getRunAtJobs());
                                } else if (Status.Running == checkSubDo.getStatus()) {
                                    checkDo.setStatus(Status.Running);
                                }
                            }
                        }
                    }
                    if (notRun != null && !notRun.isEmpty()) {
                        // 还有依赖没有执行
                        checkDo.setStatus(Status.Trigger);
                        checkDo.setRunAtJobs(notRun);
                        return checkDo;
                    }
                    if (checkDo.getStatus() == null) {
                        checkDo.setStatus(Status.Finished);
                    }
                    return checkDo;
                }

                /**
                 * 对执行中的任务，及作业依赖进行处理
                 * 
                 * @param taskRunIds
                 * @param mapTaskLogs
                 * @param mapTaskInfos
                 */
                private void runningCheck(List<Long> taskRunIds, Map<Long, TaskExecLogPo> mapTaskLogs,
                        Map<String, TaskInfoPo> mapTaskInfos) {
                    // 加载所有的job执行日志，，job是一次性写入所有的job
                    List<JobExec1LogPo> jobExec1LogPos = jobExec1LogRepository.queryListByBatchId(batchLogId,
                            taskRunIds);
                    if (jobExec1LogPos == null || jobExec1LogPos.isEmpty()) {
                        // 理论不为空
                        return;
                    }

                    // 判断是否全部完成
                    // key=bbtcodes - batchTime
                    Map<String, List<JobExec1LogPo>> jobLogsGroupCodes = new HashMap<>();
                    Map<String, Boolean> jobLogsGroupBbtStatus = new HashMap<>();
                    Map<String, Boolean> jobLogsGroupBbtjStatus = new HashMap<>();
                    for (JobExec1LogPo jobLog : jobExec1LogPos) {
                        // key= bbtcodes-batchtime
                        String bbtCode = jobLog.getBizCode() + "-" + jobLog.getBatchCode() + "-" + jobLog.getTaskCode();
                        if (jobLogsGroupCodes.containsKey(bbtCode)) {
                            jobLogsGroupCodes.get(bbtCode).add(jobLog);
                        } else {
                            List<JobExec1LogPo> jobExec1Logs = new ArrayList<>();
                            jobExec1Logs.add(jobLog);
                            jobLogsGroupCodes.put(bbtCode, jobExec1Logs);
                        }
                        String bbtCodeAndtime = bbtCode + "-" + jobLog.getBatchTriggerTime();
                        String bbtjCodeAndtime = bbtCode + "-" + jobLog.getJobCode() + "-"
                                + jobLog.getBatchTriggerTime();
                        if (!jobLogsGroupBbtStatus.containsKey(bbtCodeAndtime)) {
                            jobLogsGroupBbtStatus.put(bbtCodeAndtime, true);
                        }
                        if (!jobLogsGroupBbtjStatus.containsKey(bbtjCodeAndtime)) {
                            jobLogsGroupBbtjStatus.put(bbtjCodeAndtime, true);
                        }
                        boolean succ = false;
                        // 正常
                        if (ExecStatus.Finished.value().equals(jobLog.getExecStatus())) {
                            succ = true;
                        } else if (ExecStatus.Excption.value().equals(jobLog.getExecStatus())) {
                            // 判断task是否允许跳过
                            TaskInfoPo taskInfoPo = mapTaskInfos.get(bbtCode);
                            if (YN.Yes.value().equals(taskInfoPo.getFailIgnore())) {
                                succ = true;
                            }
                        }
                        if (!succ) {
                            if (jobLogsGroupBbtStatus.get(bbtCodeAndtime)) {
                                jobLogsGroupBbtStatus.put(bbtCodeAndtime, false);
                            }
                            if (jobLogsGroupBbtjStatus.get(bbtjCodeAndtime)) {
                                jobLogsGroupBbtjStatus.put(bbtjCodeAndtime, false);
                            }
                        }
                    }
                    // 开始检测每个tasklog执行结果
                    for (Long taskLogId : taskRunIds) {
                        TaskExecLogPo taskExecLogPo = mapTaskLogs.get(taskLogId);
                        Long batchTriggerTiem = taskExecLogPo.getBatchTriggerTime();

                        String bizCode = taskExecLogPo.getBizCode();
                        String batchCode = taskExecLogPo.getBatchCode();
                        String taskCode = taskExecLogPo.getTaskCode();
                        // 跟着依赖一层层推，，结果是当前task下的一层job
                        List<String> jobCodes = new ArrayList<>();
                        jobCodes.add(AppVariables.DEPENDON_PHD);
                        DependonCheckDo dependonCheckDo = loopJobDpdons(bizCode, batchCode, taskCode, jobCodes,
                                batchTriggerTiem, jobLogsGroupBbtjStatus, false);
                        if (Status.Running == dependonCheckDo.getStatus()) {
                            continue;
                        } else if (Status.Finished == dependonCheckDo.getStatus()) {
                            // 更新当前task 状态为 完成
                            TaskExecLogPo upTaskLogPo = new TaskExecLogPo();
                            upTaskLogPo.setId(taskLogId);
                            upTaskLogPo.setExecStatus(ExecStatus.Finished.value());
                            upTaskLogPo.setEndTime(new Date());
                            upTaskLogPo.setExecResult("");
                            taskExecLogRepository.updateById(upTaskLogPo);
                            // 通知
                            noticeHelper.recordNotice(taskExecLogPo);
                        } else {
                            // NORMAL 没有依赖触发的作业
                            // System.out.println(" not trigger  >>>" + dependonCheckDo.getRunAtJobs());
                        }
                    }
                }

                private DependonCheckDo loopJobDpdons(String bizCode, String batchCode, String taskCode,
                        List<String> jobCodes, Long batchTriggerTiem, Map<String, Boolean> jobLogsGroupBbtjStatus,
                        boolean sub) {
                    DependonCheckDo checkDo = new DependonCheckDo();
                    Set<Long> setKeys = new HashSet<>();
                    Set<DependonPo> dpdPos = new HashSet<>();
                    Set<String> notRun = new HashSet<>();

                    for (String job : jobCodes) {
                        if (AppVariables.DEPENDON_PHD.equals(job) && sub) {
                            continue;
                        }
                        List<DependonPo> dependonPos = cacheTjDpdon.getDpdsByParent(bizCode, batchCode, taskCode, job,
                                DependType.Job);
                        if (dependonPos != null) {
                            for (DependonPo dpn : dependonPos) {
                                if (!setKeys.contains(dpn.getId())) {
                                    setKeys.add(dpn.getId());
                                    dpdPos.add(dpn);
                                }
                                // 检测是否存在执行记录
                                String bbtjCode = dpn.getBizCode() + "-" + dpn.getBatchCode() + "-" + dpn.getTaskCode()
                                        + "-" + dpn.getJobCode();
                                Boolean bbtjStatus = jobLogsGroupBbtjStatus.get(bbtjCode + "-" + batchTriggerTiem);
                                if (bbtjStatus == null) {
                                    // 没有找到，说明还没有执行
                                    notRun.add(dpn.getJobCode());
                                }
                            }
                        }
                    }
                    if (dpdPos == null || dpdPos.isEmpty()) {
                        // 没有依赖，task直接完成
                        checkDo.setStatus(Status.Finished);
                        return checkDo;
                    }
                    // 优先检测完成的
                    boolean layerAllSucc = true;
                    // 从task开始查询 job
                    List<String> jcodes = new ArrayList<>();
                    for (DependonPo dpn : dpdPos) {
                        if (notRun.contains(dpn.getJobCode())) {
                            continue;
                        }
                        String bbtjCode = dpn.getBizCode() + "-" + dpn.getBatchCode() + "-" + dpn.getTaskCode() + "-"
                                + dpn.getJobCode();
                        Boolean bbtjStatus = jobLogsGroupBbtjStatus.get(bbtjCode + "-" + batchTriggerTiem);
                        // 判断下一个依赖
                        if (!bbtjStatus) {
                            layerAllSucc = false;
                        }
                        jcodes.add(dpn.getJobCode());
                    }
                    if (!layerAllSucc) {
                        checkDo.setStatus(Status.Running);
                        checkDo.setRunAtJobs(notRun);
                        return checkDo;
                    }
                    // 一层全部成功了
                    if (jcodes.size() > 0) {
                        DependonCheckDo checkSubDo = loopJobDpdons(bizCode, batchCode, taskCode, jcodes,
                                batchTriggerTiem, jobLogsGroupBbtjStatus, true);
                        if (Status.Trigger == checkSubDo.getStatus()) {
                            notRun.addAll(checkSubDo.getRunAtJobs());
                        } else if (Status.Running == checkSubDo.getStatus()) {
                            checkDo.setStatus(Status.Running);
                        }
                    }
                    if (notRun != null && !notRun.isEmpty()) {
                        // 还有依赖没有执行
                        checkDo.setStatus(Status.Trigger);
                        checkDo.setRunAtJobs(notRun);
                        return checkDo;
                    }
                    if (checkDo.getStatus() == null) {
                        checkDo.setStatus(Status.Finished);
                    }
                    return checkDo;
                }
            };

            FutureTask<Boolean> futureTask = new FutureTask<>(runner);
            listFutureTasks.add(futureTask);
            taskExecutorPool.execute(futureTask);
        }
        for (FutureTask<Boolean> futureTask : listFutureTasks) {
            try {
                futureTask.get(); // 等待所有任务完成
            } catch (Exception e) {
                e.printStackTrace();
            }
        }
    }
}
